/*
 * Copyright (C) 2025 ameise <ameise.wang@gmail.com> - All Rights Reserved
 *
 * This file is part of e3pool.
 *
 * e3pool is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * e3pool is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with e3pool. If not, see <https://www.gnu.org/licenses/>.
 */

package go_pool

import (
	"context"
	"gitee.com/ameise84/e3ds/queue"
	"gitee.com/ameise84/e3lock"
	"sync"
	"sync/atomic"
)

func newCacheRunner(h PanicHook, where string, opts Options) *runnerCache {
	r := &runnerCache{
		panicHook: h,
		where:     where,
		simChan:   make(chan struct{}, opts.simCount),
	}
	if opts.cacheSize != 0 {
		r.taskQueue = queue.NewQueueLock[*task](&e3lock.SpinLock{}, opts.cacheSize)
	} else {
		r.taskQueue = queue.NewQueueLock[*task](&e3lock.SpinLock{}, 0)
	}
	return r
}

type runnerCache struct {
	opts        Options
	panicHook   PanicHook
	where       string
	wg          sync.WaitGroup
	simChan     chan struct{}
	taskQueue   queue.IQueue[*task]
	checkWorker atomic.Int64
}

func (ts *runnerCache) SyncRun(f GoSyncFunc, args ...any) (any, error) {
	ch := make(chan taskResult)
	t := newTask(ch, f, args)
	defer freeTask(t)
	if err := ts.post(t); err != nil {
		return nil, err
	}
	rs := <-ch
	return rs.result, rs.err
}

func (ts *runnerCache) AsyncRun(f GoAsyncFunc, args ...any) error {
	t := newTask(nil, f, args)
	if err := ts.post(t); err != nil {
		freeTask(t)
		return err
	}
	return nil
}

func (ts *runnerCache) Wait() {
	ts.wg.Wait()
}

func (ts *runnerCache) post(t *task) (err error) {
	if ts.opts.limiter != nil {
		ctx := context.Background()
		if ts.opts.isBlock {
			if err = ts.opts.limiter.Wait(ctx); err != nil {
				return ErrRunnerPostTimeOut
			}
		} else {
			if !ts.opts.limiter.Allow() {
				return ErrRunnerLimit
			}
		}
	}
	ts.wg.Add(1)
	if err = ts.taskQueue.Enqueue(t); err != nil {
		ts.wg.Done()
		return ErrRunnerIsFull
	}
	select {
	case ts.simChan <- struct{}{}:
		goWorkerDo(ts.consume)
	default:
	}
	return nil
}

func (ts *runnerCache) consume(...any) {
	ts.checkWorker.Add(1)
	for {
		t, err := ts.taskQueue.Dequeue()
		if err != nil {
			break
		}
		doTaskSafe(ts.panicHook, ts.where, t)
		ts.wg.Done()
	}
	<-ts.simChan
	old := ts.checkWorker.Add(-1)
	if old == 0 && ts.taskQueue.Size() > 0 {
		goWorkerDo(ts.consume)
	}
}
